package com.stock;

import java.io.IOException;
import java.util.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

public class SentimentWordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    // 存储每个情感的前100个高频单词
    private Map<String, PriorityQueue<WordCount>> sentimentTopWords = new HashMap<>();
    private MultipleOutputs<Text, IntWritable> mos;  // 用于分文件输出

    static class WordCount {
        String word;
        int count;
        WordCount(String word, int count) {
            this.word = word;
            this.count = count;
        }
    }

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        // 初始化最小堆，分别保存正面和负面前100
        sentimentTopWords.put("1", new PriorityQueue<>(100, Comparator.comparingInt(wc -> wc.count)));
        sentimentTopWords.put("-1", new PriorityQueue<>(100, Comparator.comparingInt(wc -> wc.count)));

        // 初始化 MultipleOutputs
        mos = new MultipleOutputs<>(context);
    }

    @Override
    public void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {

        String[] parts = key.toString().split("\t");
        if (parts.length != 2) return;

        String sentiment = parts[0];
        String word = parts[1];

        // 求该词的总频次
        int sum = 0;
        for (IntWritable val : values) sum += val.get();

        PriorityQueue<WordCount> queue = sentimentTopWords.get(sentiment);
        if (queue == null) return;

        // 插入堆
        if (queue.size() < 100) {
            queue.offer(new WordCount(word, sum));
        } else if (sum > queue.peek().count) {
            queue.poll();
            queue.offer(new WordCount(word, sum));
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // 输出正面情感结果
        outputTopWords("1");
        // 输出负面情感结果
        outputTopWords("-1");

        mos.close();
    }

    private void outputTopWords(String sentiment) throws IOException, InterruptedException {
        PriorityQueue<WordCount> queue = sentimentTopWords.get(sentiment);
        if (queue == null) return;

        List<WordCount> sortedList = new ArrayList<>(queue);
        sortedList.sort((a, b) -> Integer.compare(b.count, a.count));

        for (WordCount wc : sortedList) {
            if (sentiment.equals("1")) {
                mos.write("positive", new Text(wc.word), new IntWritable(wc.count));
            } else if (sentiment.equals("-1")) {
                mos.write("negative", new Text(wc.word), new IntWritable(wc.count));
            }
        }
    }
}
